package hbase.impl

import hbase.{Executor, HBase}
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.client.{ColumnFamilyDescriptorBuilder, ConnectionFactory, Put, TableDescriptorBuilder}
import org.apache.hadoop.hbase.exceptions.HBaseException
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}

import java.util
import scala.collection.mutable.ArrayBuffer

class MyHBase (conf: Map[String, String])(pooled: Boolean = false, poolSize: Int = 3) extends HBase{
  private lazy val config: Configuration = HBaseConfiguration.create()
  private lazy val pool: ArrayBuffer[PoolCon] = ArrayBuffer()
  /**
   * 1、初始化配置信息
   * build(key:String, value: String) => statusCode 0
   */
  override def build(): HBase = {
    if(statusCode == -1){
      conf.foreach(t => config.set(t._1, t._2))
      statusCode = 0
      this
    }else{
      throw new HBaseException("build() function must be invoked first")
    }
  }
  /**
   * 2、创建连接对象
   * initPool() 初始化连接池 => statusCode 1
   */
  override def initPool(): HBase = {
    if(statusCode == 0){
      val POOL_SIZE = if (pooled) {
        if (poolSize <= 0) 3 else poolSize
      } else 1
      for (i <- 1 to POOL_SIZE) {
        pool.append(PoolCon(available = true, ConnectionFactory.createConnection(config)))
      }
      statusCode = 1
      this
    }else{
      throw new HBaseException("initPool() function must be invoked only after build()")
    }

  }

  /**
   * 3、完成 返回执行器
   * finish() 初始化连接池 => statusCode 2
   */
  override def finish(): Executor = {
    if (statusCode == 1) {
      statusCode = 2
      new Executor {

        /**
         * 验证数据表是否存在
         *
         * @param tableName 表名："命名空间:表名"
         * @return Boolean
         */
        override def exists(tableName: String): Boolean = {
          var pc: PoolCon = null
          try{
            pc = getCon
            val exists = pc.con.getAdmin.tableExists(TableName.valueOf(tableName))
            pc.in
            exists
          }catch {
            case e: Exception => e.printStackTrace()
              false
          }finally {
            close(pc)
          }
        }

        /**
         * 创建数据表
         *
         * @param tableName      表名："命名空间:表名"
         * @param columnFamilies 列族
         * @return Boolean
         */
        override def create(tableName: String, columnFamilies: Seq[String]): Boolean = {
          if (exists(tableName)) {
            return false
          }
          var pc: PoolCon = null
          try {
            pc = getCon
            val builder: TableDescriptorBuilder = TableDescriptorBuilder
              .newBuilder(TableName.valueOf(tableName))

            columnFamilies.foreach(
              cf => builder.setColumnFamily(
                ColumnFamilyDescriptorBuilder.of(cf)
              )
            )
            pc.con.getAdmin.createTable(builder.build())
            true
          } catch {
            case e: Exception => e.printStackTrace()
              false
          } finally {
              close(pc)
          }
        }

        /**
         * 删除数据表
         *
         * @param tableName 表名："命名空间:表名"
         * @return Boolean
         */
        override def drop(tableName: String): Boolean = {
          if(!exists(tableName)){
            return false
          }
          var pc: PoolCon = null
          try {
            pc = getCon
            pc.con.getAdmin.deleteTable(TableName.valueOf(tableName))
            true
          } catch {
            case e: Exception => e.printStackTrace()
              false
          } finally {
            close(pc)
          }
        }

        /**
         * 批量插入数据
         * @param tableName 表名："命名空间:表名"
         * @param data      数据
         */
        override def put(tableName: String, data: util.List[Put]): Boolean = {
          if(!exists(tableName)){
            return false
          }
          var pc: PoolCon = null
          try {
            pc = getCon
            pc.con.getTable(TableName.valueOf(tableName)).put(data)
            true
          } catch {
            case e: Exception => e.printStackTrace()
              false
          } finally {
            close(pc)
          }

        }
      }
    }
    else {
      throw new HBaseException("finish() function must be invoked only after initPool()")
    }
  }

  private def getCon = {
    val left: ArrayBuffer[PoolCon] = pool.filter(_.available)
    if (left.isEmpty) {
      throw new HBaseException("no available connection")
    }
    left.apply(0).out
  }

  private def close(con: PoolCon) = {
    if (null != con) {
      con.in
    }
  }
}

object MyHBase{
  def apply(conf: Map[String, String])(poolSize: Int): MyHBase = new MyHBase(conf)(true, poolSize)
  // def apply(conf: Map[String, String]): MyHBase = new MyHBase(conf)()
}
